package com.shujia.flink.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo3MysqlSink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.readTextFile("data/students.txt")

    //使用自定义sink
    studentsDS.addSink(new MysqlSink)

    env.execute()
  }
}

/**
 * 自定义sink将数据写入mysql
 * RichSinkFunction比SinkFunction多了open和close方法，用于创建和关闭连接
 */
class MysqlSink extends RichSinkFunction[String] {

  //提出来作为成员变量
  var stat: PreparedStatement = _
  var con: Connection = _

  /**
   * open方法，在任务启动的时候执行一次，每一个task中执行一次
   *
   */
  override def open(parameters: Configuration): Unit = {
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")
    //2、创建数据库连接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
    //3、便sql
    stat = con.prepareStatement("insert into students values(?,?,?,?,?)")

  }

  //在任务结束时执行，用于关闭连接
  override def close(): Unit = {
    //6、关闭数据库连接
    con.close()
  }

  /**
   * invoke:每一条数据执行一次
   *
   * @param stu     ：一行数据
   * @param context ：上下文对象
   */
  override def invoke(stu: String, context: SinkFunction.Context): Unit = {

    //4、赋值
    val split: Array[String] = stu.split(",")
    stat.setString(1, split(0))
    stat.setString(2, split(1))
    stat.setInt(3, split(2).toInt)
    stat.setString(4, split(3))
    stat.setString(5, split(4))
    //5、只sl
    stat.execute()
  }
}